package ar.uba.fi.tonyvaliente.tree;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.logging.Logger;

import ar.uba.fi.tonyvaliente.files.Persistor;
import ar.uba.fi.tonyvaliente.files.PersistorException;
import ar.uba.fi.tonyvaliente.files.RecordPosition;

/**
 * Persistor para objetos BTreeNode.
 * Debe estar asociado a un BTreeFileWrapper para poder deserializar
 * las referencias a otros nodos.
 * Realiza la serializacion/deserializacion de claves y valores a traves de
 * otros objetos Persistor que recibe al ser instanciado.
 * 
 * @author Diego Morello
 *
 * @param <K> Clase a la cual pertenecen las claves del nodo
 * @param <V> Clase a la cual pertenecen los valores del nodo
 */
class BTreeNodePersistor<K extends Comparable<K>, V> extends Persistor<BTreeNode<K,V>> {
	
	private static Logger log = Logger.getLogger(NodePersistor.class.getName());
	
	private Persistor<K> keyPersistor;
	private Persistor<V> valuePersistor;
	private BTreeFileWrapper<K,V> associatedWrapper;
	
	/**
	 * Crea una nueva instancia de la clase, con un Persistor para claves y uno
	 * para valores, ademas de un file wrapper al cual se asocia para poder
	 * deserializar referencias a nodos.
	 * 
	 * @author Diego Morello
	 * 
	 * @param keyPersistor Persistor para claves
	 * @param valuePersistor Persistor para valores
	 * @param wrapper File wrapper asociado
	 * @throws NullPointerException Si alguno de los parametros es null
	 */
	public BTreeNodePersistor(Persistor<K> keyPersistor, Persistor<V> valuePersistor,
			BTreeFileWrapper<K,V> wrapper) {
		
		if (keyPersistor == null || valuePersistor == null || wrapper == null) {
			throw new NullPointerException("Error al inicializar NodePersistor");
		}
		
		this.keyPersistor = keyPersistor;
		this.valuePersistor = valuePersistor;
		this.associatedWrapper = wrapper;
	}
	
	/**
	 * Serializa un objeto BTreeNode.
	 * La serializacion de claves y valores se hace a traves de los Persistors
	 * con los cuales fue instanciado.
	 * 
	 * Estructura del nodo serializado:
	 * 
	 * - cantidad de elementos ................ 4 bytes
	 * - cantidad maxima de elementos ......... 4 bytes
	 * - numero de bloque del nodo padre ...... 4 bytes
	 * - offset del nodo padre en el bloque ... 4 bytes
	 * - cantidad de elementos del arbol ...... 4 bytes
	 * --------
	 * Por cada elemento del arbol:
	 * - cantidad de bytes de la clave ........ 4 bytes
	 * - bytes de la clave .................... *
	 * - cantidad de bytes del valor .......... 4 bytes
	 * - bytes del valor ...................... *
	 * --------
	 * - cantidad de referencias a hijos ...... 4 bytes
	 * Por cada referencia a un nodo hijo:
	 * - numero de bloque del nodo hijo ....... 4 bytes
	 * - offset del nodo hijo en el bloque .... 4 bytes
	 * 
	 * @author Diego Morello
	 * 
	 * @param node Nodo a serializar
	 * @return byte array con el nodo serializado
	 * @throws PersistorException si ocurre un error al serializar
	 */
	public byte[] serialize(BTreeNode<K,V> node) throws PersistorException {
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		DataOutputStream out = new DataOutputStream(baos);
		try {
			// Escribo los "datos de cabecera" del nodo
			int cantElementos = node.getCantidadElementos();
			int maxCantElementos = node.getMaxCantidadElementos();
			
			out.writeInt(cantElementos);
			out.writeInt(maxCantElementos);
			
			// Escribo la referencia al nodo padre
			DiskBReference<K,V> parent = (DiskBReference<K,V>)node.getParent();
			this.writeDiskBReference(parent, out);
			
			// Escribo cantidad de elementos del arbol
			Map<K,V> datos = node.getDatos();
			out.writeInt(datos.size());
			
			// Escribo datos del arbol
			Iterator<K> it = datos.keySet().iterator();
			while (it.hasNext()) {
				K key = it.next();
				this.writeKey(key, out);
				this.writeValue(datos.get(key), out);
			}
			
			// Escribo la cantidad de referencias
			List<BReference<K, V>> references = node.getReferences();
			out.writeInt(references.size());
			
			for (BReference<K,V> ref: references) {
				this.writeDiskBReference((DiskBReference<K,V>)ref, out);
			}
			
			out.close();
		} catch (IOException e) {
			throw new PersistorException("Error al serializar BTreeNode", e);
		}
		return baos.toByteArray();
	}
	
	/**
	 * Crea un BTreeNode a partir del byte array recibido por parametro.
	 * La deserializacion de claves y valores se hace a traves de los Persistors
	 * con los cuales fue instanciado.
	 *
	 * Estructura del nodo serializado:
	 * 
	 * - cantidad de elementos ................ 4 bytes
	 * - cantidad maxima de elementos ......... 4 bytes
	 * - numero de bloque del nodo padre ...... 4 bytes
	 * - offset del nodo padre en el bloque ... 4 bytes
	 * - cantidad de elementos del arbol ...... 4 bytes
	 * --------
	 * Por cada elemento del arbol:
	 * - cantidad de bytes de la clave ........ 4 bytes
	 * - bytes de la clave .................... *
	 * - cantidad de bytes del valor .......... 4 bytes
	 * - bytes del valor ...................... *
	 * --------
	 * - cantidad de referencias a hijos ...... 4 bytes
	 * Por cada referencia a un nodo hijo:
	 * - numero de bloque del nodo hijo ....... 4 bytes
	 * - offset del nodo hijo en el bloque .... 4 bytes
	 * 
	 * @author Diego Morello
	 * 
	 * @param bytes byte array con el nodo serializado
	 * @return BTreeNode deserializado
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	public BTreeNode<K,V> deserialize(byte[] bytes) throws PersistorException {
		ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
		DataInputStream in = new DataInputStream(bais);
		
		BTreeNode<K,V> node = null;
		
		int cantElementos;
		int maxCantElementos;
		DiskBReference<K,V> parent;
		TreeMap<K,V> datos;
		ArrayList<BReference<K,V>> references;
		
		try {
			// Leo la cantidad de elementos
			cantElementos = in.readInt();
			if (cantElementos < 0) {
				throw new PersistorException("Error al deserializar nodo: "
						+ "la cantidad de elementos no puede ser negativa");
			}
			
			// Leo la cantidad maxima de elementos
			maxCantElementos = in.readInt();
			if (maxCantElementos < 0) {
				throw new PersistorException("Error al deserializar nodo: "
						+ "la cantidad maxima de elementos no puede ser negativa");
			}
			
			// Leo la referencia al nodo padre
			parent = this.readDiskBReference(in);
			
			// Leo la cantidad de pares clave-valor
			int size = in.readInt();
			if (size < 0) {
				throw new PersistorException("Error al deserializar nodo: "
						+ "la cantidad de elementos no puede ser negativa");
			}
			
			// Leo claves y valores
			datos = new TreeMap<K,V>();
			for (int i = 0; i < size; i++) {
				K key = this.readKey(in);
				V value = this.readValue(in);
				datos.put(key, value);
			}
			
			// Leo cantidad de referencias
			int cantRef = in.readInt();
			if (cantRef < 0) {
				throw new PersistorException("Error al deserializar nodo: "
						+ "la cantidad de referencias no puede ser negativa");
			}
			
			// Leo las referencias
			references = new ArrayList<BReference<K,V>>();
			for (int i = 0; i < cantRef; i++) {
				DiskBReference<K,V> ref = this.readDiskBReference(in);
				references.add(ref);
			}
			
			in.close();
			
			node = new BTreeNode<K,V>(parent, maxCantElementos);
			node.setCantidadElementos(cantElementos);
			node.setDatos(datos);
			node.setReferences(references);
			
		} catch (IOException e) {
			throw new PersistorException("Error al deserializar nodo", e);
		}
		
		return node;
	}
	
	/**
	 * Escribe una DiskReference en el DataOutputStream recibido
	 * @param ref referencia a escribir
	 * @param out DataOutputStream donde escribir la referencia
	 * @throws PersistorException si ocurre algun error al escribir
	 */
	private void writeDiskBReference(DiskBReference<K,V> ref, DataOutputStream out) throws PersistorException {
		try {
			if (ref != null) {
				out.writeInt(ref.getRecordPosition().getBlockNumber());
				out.writeInt(ref.getRecordPosition().getOffset());
			} else {
				out.writeInt(-1);
				out.writeInt(-1);
			}
		} catch (IOException e) {
			throw new PersistorException("Error al serializar referencia", e);
		}
	}
	
	/**
	 * Lee una DiskReference del DataInputStream recibido. 
	 * @param in
	 * @return objeto DiskReference
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	private DiskBReference<K,V> readDiskBReference(DataInputStream in)
			throws PersistorException {
		
		try {
			int blockNumber = in.readInt();
			int offset = in.readInt();
		
			if (blockNumber >= 0 && offset >= 0) {
				DiskBReference<K,V> ref = new DiskBReference<K,V>(associatedWrapper);
				ref.setRecordPosition(new RecordPosition(blockNumber, offset));
				return ref;
			} else if (blockNumber < 0 && offset < 0) {
				return null;
			} else {
				throw new PersistorException("Error al deserializar referencia a nodo: "
						+ "la posicion leida no es valida");
			}
		} catch (IOException e) {
			throw new PersistorException("Error al deserializar referencia a nodo", e);
		}
	}
	
	/**
	 * Escribe una clave en el DataOutputStream recibido por parametro
	 * @param key clave a escribir
	 * @param out DataOutputStream donde escribir
	 * @throws PersistorException si ocurre un error al escribir
	 */
	private void writeKey(K key, DataOutputStream out) throws PersistorException {
		try {
			byte[] keyBytes = keyPersistor.serialize(key);
			out.writeInt(keyBytes.length);
			out.write(keyBytes);
		} catch (IOException e) {
			throw new PersistorException("Error al escribir clave");
		}
	}
	
	/**
	 * Deserializa una clave a partir del DataInputStream recibido
	 * @param in
	 * @return clave (objeto de clase K)
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	private K readKey(DataInputStream in) throws PersistorException {
		try {
			int keySize = in.readInt();
			if (keySize >= 0) {
				byte[] keyBytes = new byte[keySize];
				if (in.read(keyBytes) != keySize) {
					throw new PersistorException("Error al deserializar clave: "
						+ "numero erroneo de bytes leidos");
				}
				return keyPersistor.deserialize(keyBytes);
			} else {
				throw new PersistorException("Error al deserializar clave: "
						+ "el tamanio en bytes no puede ser negativo");
			}
		} catch (IOException e) {
			throw new PersistorException("Error al deserializar clave");
		}
	}
	
	/**
	 * Escribe un valor en el DataOutputStream recibido por parametro
	 * @param value valor a escribir
	 * @param out DataOutputStream donde escribir
	 * @throws PersistorException si ocurre un error al escribir
	 */
	private void writeValue(V value, DataOutputStream out) throws PersistorException {
		try {
			byte[] valueBytes = valuePersistor.serialize(value);
			out.writeInt(valueBytes.length);
			out.write(valueBytes);
		} catch (IOException e) {
			throw new PersistorException("Error al escribir valor");
		}
	}
	
	/**
	 * Deserializa un valor a partir del DataInputStream recibido
	 * @param in
	 * @return valor (objeto de clase V)
	 * @throws PersistorException si ocurre un error al deserializar
	 */
	private V readValue(DataInputStream in) throws PersistorException {
		try {
			int valueSize = in.readInt();
			if (valueSize >= 0) {
				byte[] valueBytes = new byte[valueSize];
				if (in.read(valueBytes) != valueSize) {
					throw new PersistorException("Error al deserializar valor: "
						+ "numero erroneo de bytes leidos");
				}
				return valuePersistor.deserialize(valueBytes);
			} else {
				throw new PersistorException("Error al deserializar valor: "
						+ "el tamanio en bytes no puede ser negativo");
			}
		} catch (IOException e) {
			throw new PersistorException("Error al deserializar valor");
		}
	}
}
